#include "EventEngine.h"
#include <cstdint>
#include <functional>
#include <iostream>
#include <memory>
#include <regex>
#include <sstream>
#include <string>
#include <utility>
#include "DateTime.h"
#include "EventLoop.h"
#include "InnerType.h"
#include "LoopManager.h"
#include "Manager.h"
#include "MessageHeader.h"
#include "Poller.h"
#include "SocketOpt.h"
#include "StreamServer.h"
#include "Util.h"
#include "XnccType.h"
namespace xncc {
namespace user {

    template <typename Ret, typename... Args>
    static void function_params_check(std::function<Ret(Args...)> arg)
    {
        static_assert(sizeof...(Args) == 4, "Args mutst is four");
    };

    using namespace msg;
    using namespace types;
    using namespace shared;

    static const std::string REGSTR =
        "^((([0-9]{3}[1-9]|[0-9]{2}[1-9][0-9]{1}|[0-9]{1}[1-9][0-9]{2}|[1-9][0-9]{3})-(((0[13578]|1["
        "02])-(0[1-9]|[12][0-9]|3[01]))|((0[469]|11)-(0[1-9]|[12][0-9]|30))|(02-(0[1-9]|[1][0-9]|2[0-"
        "8]))))|((([0-9]{2})(0[48]|[2468][048]|[13579][26])|((0[48]|[2468][048]|[3579][26])00))-02-"
        "29))_([0-1][0-9]|2[0-3]):([0-5][0-9]):([0-5][0-9]).([0-9][0-9][0-9])$";
    class EventEngineImpl
    {
        using ConnEventCallBack = std::function<void(connection_id_t, thread_id_t)>;

      private:
        foundation::EventLoop*                                                loop_;
        std::regex                                                            datePattern_;
        ConnEventCallBack                                                     connCallback_;
        std::function<void(connection_id_t, thread_id_t, const std::string&)> disconnCallback_;
        engine_id_t                                                           id_;
        ska::flat_hash_map<connection_id_t, std::string>                      connAddrs_;

      public:
        explicit EventEngineImpl(engine_id_t engine_id) : loop_(LOOP_MGR.getLoop(engine_id).get()), id_(engine_id)
        {
            if (loop_ == nullptr) {
                std::ostringstream oss;
                oss << " EventEngine id = " << engine_id << " not found in conf.json";
                throw logic_exception(oss.str());
            }
        }
        EventEngineImpl(const EventEngineImpl&)             = delete;
        EventEngineImpl(const EventEngineImpl&&)            = delete;
        EventEngineImpl& operator=(const EventEngineImpl&)  = delete;
        EventEngineImpl& operator=(const EventEngineImpl&&) = delete;

        ~EventEngineImpl() {}
        [[nodiscard]] engine_id_t id() const { return id_; }

        std::string getPeerAddr(types::connection_id_t conn_id) { return connAddrs_[conn_id]; }

        void setConnCallback(const ConnEventCallBack& user_func)
        {
            connCallback_ = user_func;
            loop_->registerUserFunc(InnerMsgId::ON_CONNECTED,
                                    [this](const char* buf, uint32_t len, thread_id_t tid, connection_id_t connId) {
                                        handleConnected(buf, len, tid, connId);
                                    });
        }
        void setDisconnCallback(std::function<void(connection_id_t, thread_id_t, const std::string&)> user_func)
        {
            disconnCallback_ = std::move(user_func);
            loop_->registerUserFunc(InnerMsgId::ON_DISCONNECTED,
                                    [this](const char* buf, uint32_t len, thread_id_t tid, connection_id_t connId) {
                                        handleDisconnected(buf, len, tid, connId);
                                    });
        }
        void registerUserFunc(message_id_t msg_id, const UserRegFunc& user_func)
        {
            if (msg_id <= InnerMsgId::MAX_INNER_MSG_ID) {
                throw logic_exception(" register id must great than " + std::to_string(InnerMsgId::MAX_INNER_MSG_ID));
            }
            loop_->registerUserFunc(msg_id, user_func);
        }

        timer_id_t setTimerFunc(int64_t milli, int64_t interval, const Task& task)
        {
            return loop_->runAt(milli, task, interval);
        }

        void handleConnected(const char* buf, uint32_t len, thread_id_t tid, connection_id_t connId)
        {
            connAddrs_[connId] = std::string(buf, len);
            connCallback_(connId, tid);
        }
        void handleDisconnected(const char* buf, uint32_t len, thread_id_t tid, connection_id_t connId)
        {
            disconnCallback_(connId, tid, std::string(buf, len));
            connAddrs_.erase(connId);
        }

        static bool kickConnection(types::connection_id_t connId, types::thread_id_t tid)
        {
            // 业务线程调用 IO线程执行
            auto tcpLoop = LOOP_MGR.getIpcLoop(tid);  // 运行中不发生变化  没有竞争问题
            if (tcpLoop == nullptr) {
                return false;
            }
            tcpLoop->runInLoop([tcpLoop, connId]() { tcpLoop->poller()->kickInLoop(connId); });
            return true;
        }

        timer_id_t runAtOnce(const std::string& tpoint, const Task& task)
        {
#if 0      
//gcc4.8.5对正则支持有限    
            bool formatOk = std::regex_match(tpoint, datePattern_);
            if (!formatOk) {
                return INVALID_TIMER_ID;
            }
#endif
            int64_t now = DateTime::timeMicro();
            std::tm t   = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0};
            int64_t ms;
            int     ret = std::sscanf(tpoint.data(), "%04d-%02d-%02d_%02d:%02d:%02d.%03ld", &t.tm_year, &t.tm_mon,
                                      &t.tm_mday, &t.tm_hour, &t.tm_min, &t.tm_sec, &ms);
            --t.tm_mon;
            t.tm_year -= types::DateTime::YEAR_1900;
            if (ret != 7) {
                return INVALID_TIMER_ID;
            }
            std::time_t expectedMicro = std::mktime(&t);
            if (expectedMicro == -1) {
                return INVALID_TIMER_ID;
            }
            expectedMicro = expectedMicro * DateTime::MICROS_PER_SEC + ms * DateTime::MILLS_PER_SEC;
            if (now >= expectedMicro) {
                return INVALID_TIMER_ID;
            }
            return loop_->runAt(expectedMicro, task, 0);
        }

        void cancelTimer(timer_id_t timer_id) { loop_->cancel(timer_id); }

        int subQueue(queue_id_t queueId, queue_size_t queueSize, bool autoExpand)
        {
            return LOOP_MGR.subQueue(loop_->id(), queueId, queueSize, autoExpand);
        }

        static int dynamicSendData(const char* buf, uint32_t len, message_id_t msg_id, queue_id_t queueId)
        {
            return QUEUE_MGR.dynamicPubTreadData(buf, len, msg_id, queueId);
        }

        static int dynamicSendDataWithLock(const char* buf, uint32_t len, message_id_t msg_id, queue_id_t queueId)
        {
            return QUEUE_MGR.dynamicPubTreadDataWithLock(buf, len, msg_id, queueId);
        }

        int sendData(const char* buf, uint32_t len, message_id_t msg_id)
        {
            return QUEUE_MGR.pubThreadData(buf, len, msg_id, loop_->id());
        }

        int sendData(const char* msgBody, uint32_t len, message_id_t msg_id, thread_id_t tid, connection_id_t connId)
        {
            return QUEUE_MGR.pubMsgToIPCThread(msgBody, len, tid, loop_->id(), connId, msg_id);
        }

        [[nodiscard]] const std::vector<types::queue_id_t>& GetPubIds() const
        {
            return QUEUE_MGR.GetPubIds(loop_->id());
        }

        int sendAny(const char* buf, uint32_t len, types::message_id_t msg_id, types::queue_id_t queueId)
        {
            return QUEUE_MGR.pubThreadData(buf, len, msg_id, loop_->id(), queueId);
        }
    };

    EventEngine::EventEngine(int engine_id) : inUse_(new EventEngineImpl(engine_id))
    {
        inUse_->setConnCallback([this](connection_id_t connId, thread_id_t tid) { handleConnected(connId, tid); });
        inUse_->setDisconnCallback(
            [this](connection_id_t connId, thread_id_t tid, const std::string& disconnectedReason) {
                handleDisconnected(connId, tid, disconnectedReason);
            });
    }

    types::engine_id_t EventEngine::id() { return inUse_->id(); }

    EventEngine::~EventEngine()
    {
        if (inUse_ != nullptr) {
            delete inUse_;
        }
    }

    bool EventEngine::kickConnection(types::connection_id_t conn, types::thread_id_t tid)
    {
        return xncc::user::EventEngineImpl::kickConnection(conn, tid);
    }

    void EventEngine::registerUserFunc(message_id_t msg_id, const UserRegFunc& user_func)
    {
        function_params_check(user_func);
        inUse_->registerUserFunc(msg_id, user_func);
    }

    timer_id_t EventEngine::setTimerFunc(int64_t milli, int64_t interval, const Task& task)
    {
        return inUse_->setTimerFunc(milli, interval, task);
    }

    timer_id_t EventEngine::setIntervalFuncFromNow(int64_t interval, const Task& task)
    {
        if (interval <= 0) {
            return INVALID_TIMER_ID;
        }
        return inUse_->setTimerFunc(DateTime::timeMicro(), interval, task);
    }
    void EventEngine::cancelTimer(timer_id_t timer_id) { inUse_->cancelTimer(timer_id); }

    int EventEngine::subQueue(queue_id_t queueId, queue_size_t queueSize, bool autoExpand)
    {
        return inUse_->subQueue(queueId, queueSize, autoExpand);
    }

    int EventEngine::sendData(const char* buf, uint32_t len, message_id_t msg_id, queue_id_t queueId)
    {
        return EventEngineImpl::dynamicSendData(buf, len, msg_id, queueId);
    }

    int EventEngine::sendDataWithLock(const char* buf, uint32_t len, message_id_t msg_id, queue_id_t queueId)
    {
        return EventEngineImpl::dynamicSendDataWithLock(buf, len, msg_id, queueId);
    }

    int EventEngine::sendData(const char* buf, uint32_t len, message_id_t msg_id)
    {
        return inUse_->sendData(buf, len, msg_id);
    }

    int EventEngine::sendAny(const char* buf, uint32_t len, types::message_id_t msg_id, types::queue_id_t queueId)
    {
        return inUse_->sendAny(buf, len, msg_id, queueId);
    }

    const std::vector<types::queue_id_t>& EventEngine::GetPubIds() const { return inUse_->GetPubIds(); }

    int
    EventEngine::sendData(const char* buf, uint32_t len, message_id_t msg_id, thread_id_t tid, connection_id_t connId)
    {
        return inUse_->sendData(buf, len, msg_id, tid, connId);
    }

    timer_id_t EventEngine::runAtOnce(const std::string& tpoint, const Task& task)
    {
        return inUse_->runAtOnce(tpoint, task);
    }

    std::string EventEngine::getPeerAddr(types::connection_id_t conn_id) { return inUse_->getPeerAddr(conn_id); }
}  // namespace user
}  // namespace xncc